package com.foreveross.proxyip.core.pools.thread.chan;

import java.util.HashSet;
import java.util.List;

import com.foreveross.proxyip.core.BaseCode;
import com.foreveross.proxyip.core.pools.ChannelPool;
import com.foreveross.proxyip.core.pools.thread.AbsCheckInThread;
import com.foreveross.proxyip.core.redis.RedisExecutor;
import com.foreveross.proxyip.core.redis.RedisLock;
import com.foreveross.proxyip.core.redis.RedisManager;
import com.foreveross.proxyip.core.zk.task.strategy.CheckInTask;
import com.foreveross.proxyip.entity.ProxyIpBean;
import com.foreveross.proxyip.infra.conf.PropertyUtils;
import com.foreveross.proxyip.infra.conf.system.DefaultPropertiesLoader;

public class ChannelCheckInThread extends AbsCheckInThread {
	
	private boolean isOpenIpUniqueness = PropertyUtils.getBooleanProperty("proxyip.isOpenIpUniqueness", DefaultPropertiesLoader.CONFIG_FILE);
	private boolean isOpenChannelPoolCheckInVerify = PropertyUtils.getBooleanProperty("proxyip.isOpenChannelPoolCheckInVerify", DefaultPropertiesLoader.CONFIG_FILE);

	
	public ChannelCheckInThread(ChannelPool pool) throws Exception {
		super(pool, CheckType.CHECKIN);
		super.separationTime = Integer.parseInt(BaseCode.getValue("CHANNEL_IN_SEPARATION_TIME"));
	}

	@Override
	public boolean loadSourceIps(Long currSize) throws Exception{
		boolean result = false;
		String sourceKey = CheckInTask.getSourceKey(this.pool);
		ChannelPool cpool = (ChannelPool)this.pool;
		List<String> ips  = cpool.publicPool.poolIpMap.getKeys(0L, -1L);
		long start = System.currentTimeMillis();
		long spendTime = 0;
		
		if (ips == null) {
			return result;
		}
		
		for (final String ip : new HashSet<String>(ips)) {
			if (isOpenIpUniqueness) {
				if (this.pool.isAlreadyDistributionChannel(ip)) {
					continue;
				}
			} else {
				if (this.pool.contains(ip)) {
					continue;
				}
			}
			
			// 开启检入校验
			if (isOpenChannelPoolCheckInVerify) {
				RedisManager.getInstance().lPush(sourceKey, ip);//任务放入队列
				RedisManager.getInstance().len(sourceKey);
				result = true;
				
			// 否则直接进入渠道池中
			} else {
				new RedisExecutor<Long>() {
					@Override
					public Long run() throws Exception {
						tran = jedis.multi();
						
						ProxyIpBean bean = (ProxyIpBean) pool.ipStoreMap.get(ip);
						if(bean == null) {
							return null;
						}
						
						pool.set(tran, bean.getChannelUseCount().get(pool.poolSign), bean);
						tran.expire(ip.getBytes(), 360000);
						tran.exec();
						return null;
					}
				}.exec();
			}
		}
		
		spendTime = System.currentTimeMillis() - start;
		
		if (spendTime > 10000) {
			log.warn("池【{}】载入{}个ip耗时{}毫秒，大于默认10000", pool.poolSign, currSize, spendTime);
		}
		
		return result;
	}

	@Override
	public void publishTask() throws Exception {
		if (isOpenChannelPoolCheckInVerify) {
			super.publishTask();
			return;
		}
		
		RedisLock lock = new RedisLock(this.pool.poolSign + "_CheckInTasklock", 60 * 5);// 需要加锁，不能并行。
		Long currSize = null;
		Long begin = null;
		Long needNum = null;
		
		try {
		
			// 如果上锁成功，则执行下面操作否则放弃此次操作
			if (!lock.tryLock()) {
				return;
			}
		
			currSize = this.pool.size();
			currSize = currSize == null ? 0L : currSize;
			
			if (pool.getProxyMin() <= currSize) {
				return;
			}
			
			begin = System.currentTimeMillis();
			needNum = getNeedSize(currSize);
			this.loadSourceIps(needNum);
			
			log.debug("无验证渠道池【{}】准备检入{}个ip，执行总耗时{}毫秒", pool.poolSign, needNum, System.currentTimeMillis() - begin);
		} finally {
			lock.unlock();
		}

	}
}
